 /* 
 */
package config;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;

import mapper.GraphMapper;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputLogFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import reducer.GraphReducer;
import util.ArrayUtil;

/**
 * @author puneet
 * 
 * 
 */
public class ShortestPathJobRunner {

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		Path sparseMatrixFileLocation = new Path(args[0]);
		if (!fs.exists(sparseMatrixFileLocation)) {
			throw new IOException(
					"File : "
							+ args[0]
							+ " does not exist is HDFS. Please put the file on HDFS first");
		}
		FSDataInputStream in = fs.open(sparseMatrixFileLocation);

		BufferedReader br = new BufferedReader(new InputStreamReader(in));
		Map<String, String[]> nodeNeighborMap = new HashMap<String, String[]>();
		String s = null;
		while ((s = br.readLine()) != null) {
			String[] vals = s.split("\\s");
			nodeNeighborMap
					.put(vals[0], ArrayUtil.splice(vals, 1, vals.length));
		}

		in.close();
		br.close();

		String source = args[1];
		String dest = args[2];

		if (!nodeNeighborMap.containsKey(source)
				|| !nodeNeighborMap.containsKey(dest)) {
			throw new Exception(
					"Source or Destnation does not exist in the i/p file");
		}
		Path tempMapFilePath = new Path("tmp/mapinput");
		if (fs.exists(tempMapFilePath)) {
			fs.delete(tempMapFilePath, false);
		}
		FSDataOutputStream out = fs.create(tempMapFilePath);
		String tempFileContents = source + " 0 "
				+ ArrayUtil.join(nodeNeighborMap.get(source), " ") + "\n";
		out.write(tempFileContents.getBytes());
		out.close();
		Configuration jobConfig = new Configuration();
		Job job = new Job(jobConfig, "graph search map reduce");
		job.setMapperClass(GraphMapper.class);
		job.setReducerClass(GraphReducer.class);
		job.setJarByClass(ShortestPathJobRunner.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		MyInputFormat.addInputPath(job, tempMapFilePath);
		
		Path tempOutputDir = new Path("tmp/op/");
		if (fs.exists(tempOutputDir)) {
			fs.delete(tempOutputDir, true);
		}
		FileOutputFormat.setOutputPath(job, tempOutputDir);
		job.waitForCompletion(true);
		int i=0;
		while (true) {
			System.out.println("Running for:" + (++i));
			Path[] outputFileList = FileUtil.stat2Paths(fs.listStatus(tempOutputDir,new OutputLogFilter()));
			
			tempMapFilePath = new Path("tmp/mapinput");
			if (fs.exists(tempMapFilePath)) {
				fs.copyToLocalFile(tempMapFilePath, new Path("lastmapinput"));
				fs.delete(tempMapFilePath, false);
			}
			out = fs.create(tempMapFilePath);
			for(Path opfile:outputFileList) {
				br = new BufferedReader(new InputStreamReader(fs.open(opfile)));
				while((s=br.readLine()) != null) {
					//System.out.println("Line = " + s);
					String node = s.split("\\s")[0];
					if(node.equals(dest)) {
						System.out.println("Shortest Distance Found:"+s.split("\\s")[1]);
						out.close();
						return;
					}
					String toWrite =s+"\t"+ArrayUtil.join(nodeNeighborMap.get(s.split("\\s")[0]), "\t") + "\n"; 
					out.write(toWrite.getBytes());
				}
			}
			out.close();
			fs.delete(tempOutputDir,true);
			jobConfig = new Configuration();
			job = new Job(jobConfig, "graph search map reduce");
			job.setMapperClass(GraphMapper.class);
			
			job.setReducerClass(GraphReducer.class);
			job.setJarByClass(ShortestPathJobRunner.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);

			MyInputFormat.addInputPath(job, tempMapFilePath);
			tempOutputDir = new Path("tmp/op/");
			if (fs.exists(tempOutputDir)) {
				fs.delete(tempOutputDir, true);
			}
			FileOutputFormat.setOutputPath(job, tempOutputDir);
			job.waitForCompletion(true);			
		}
		// Now make the first file
		// Configuration config = new Configuration();
		// Job job = new Job(config, "graph map reduce");
		// job.setJarByClass(JobRunner.class);
		// job.setMapperClass(GraphMapper.class);
		// job.setReducerClass(GraphReducer.class);
		// job.setOutputKeyClass(Text.class);
		// job.setOutputValueClass(IntWritable.class);
		// TextInputFormat.addInputPath(job, new Path(args[0]));
		// FileOutputFormat.setOutputPath(job, new Path(args[1]));
		// System.exit(job.waitForCompletion(true) ? 0 : 1);

	}
	

}
